<?php
/* vim: set expandtab tabstop=4 shiftwidth=4 encoding=utf-8: */
/* $Id: */
/*
Copyright (c) 2010, Justin Swanhart
All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
    * Redistributions of source code must retain the above copyright
      notice, this list of conditions and the following disclaimer.
    * Redistributions in binary form must reproduce the above copyright
      notice, this list of conditions and the following disclaimer in the
      documentation and/or other materials provided with the distribution.
    * Neither the name of the <organization> nor the
      names of its contributors may be used to endorse or promote products
      derived from this software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY
DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/

error_reporting(E_ALL);
require_once('common.php');

interface ShardKeyMapper {
        function __construct($config, $no_change_db);
        function create_store();
        function get_shards($schema_name = null);
        function add_shard($shard_name, $shard, $coord_shard = true, $accepts_new_rows = true, $extra_info = "NULL", $schema_name = null);
        function refresh_shard_cache($schema_name = null);
        function remove_shard($shard_name, $schema_name = null);
        function drop_store();
        function add_schema($schema_name, $is_default_schema = true);
        function add_column($column, $schema_name = null,$datatype = 'integer',$next_value=1);
        function move_key($column, $key, $tables, $dest_shard, $schema_name = null );
        function lock_key($column, $key, $lock_mode, $schema_name = null);
        function unlock_key($column, $key, $schema = null); 
        function get_column_info($column_name, $schema_name = null);
        function new_key($column, $key = null, $shard_name = null, $schema_name = null, $col_info = null);
        function &map($column, $key, $schema_name=null, $operator = "=", $col_info  = null);
        function &get_keys($column, $shard_name, $schema_name = null, $col_info = null);
        function &get_key_distribution($column, $schema_name = null);
        function get_column($shema_name = null);

}


class DirectoryShardKeyMapper implements ShardKeyMapper{
	var $shard_cache = array();
	var $conn = null;  

	#The constructor accepts an array which includes host, user, password and db
	public function __construct($config, $no_change_db = false) {

		if(empty($config['host']) || empty($config['user']) || empty($config['db'])) {
			throw new Exception('Missing one of required options: host, user or db');
		}

		$this->config = $config;

		$this->connect($config, $no_change_db);
	} 
	
	private function connect($config, $no_change_db = false) {
		if($this->conn) return $this->conn;
		$this->conn = SimpleDAL::factory($config);
		
		if($this->conn->my_error()) throw new Exception($this->conn->my_error());
	
		if(!$no_change_db) { 
			$this->conn->my_select_db($this->config['db']);
			if($this->conn->my_error()) throw new Exception($this->conn->my_error());
		}
	}

	private function execute($sql, $check_for_errors = true) {
		$stmt = $this->conn->my_query($sql);
	
		if($check_for_errors) {
			if(!$stmt) throw new Exception('Error while running:' . "\n$sql\n" . $this->conn->my_error());
		}

		return $stmt;
	}

    public function my_insert_id() {
        return $this->conn->my_insert_id();
    }


	public function escape($val) {
		return $this->conn->my_escape_string($val);
	}
	
	#create the shard data store online (this is used for initial setup)
	public function create_store() {

		if(!$this->changedb($this->config['db'],true)) {
			$this->execute("CREATE DATABASE `" . $this->config['db'] . "`");
			$this->conn->my_select_db($this->config['db']);
		}

$sql = "
CREATE TABLE `schemata` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `schema_name` varchar(50) NOT NULL,
  `is_default_schema` tinyint(1) NOT NULL DEFAULT '0',
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=10 DEFAULT CHARSET=latin1;
";
		echo "Creating table: schemata\n";
		$this->execute($sql);



$sql = "
CREATE TABLE `shards` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `schema_id` int(11) NOT NULL,
  `shard_name` varchar(50) NOT NULL,
  `extra_info` text,
  `shard_rdbms` enum('mysql','pgsql') NOT NULL DEFAULT 'mysql',
  `coord_shard` tinyint(1) NOT NULL DEFAULT '1',
  `accepts_new_rows` tinyint(1) NOT NULL DEFAULT '1',
  `username` varchar(64) DEFAULT NULL,
  `password` varchar(255) DEFAULT NULL,
  `host` varchar(255) DEFAULT NULL,
  `port` smallint(6) DEFAULT NULL,
  `db` varchar(45) DEFAULT NULL,
  `enabled` tinyint(1) DEFAULT '1',
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `schema_id` (`schema_id`,`shard_name`),
  CONSTRAINT `shards_ibfk_1` FOREIGN KEY (`schema_id`) REFERENCES `schemata` (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=86 DEFAULT CHARSET=latin1;
";
		echo "Creating table: shards\n";
		$this->execute($sql);

$sql = "
CREATE TABLE `schemata_config` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `schema_id` int(11) NOT NULL,
  `var_name` varchar(20) NOT NULL,
  `var_value` text NOT NULL,
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `schema_id` (`schema_id`,`var_name`),
  CONSTRAINT `schemata_config_ibfk_1` FOREIGN KEY (`schema_id`) REFERENCES `schemata` (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=26 DEFAULT CHARSET=latin1;
";
		echo "Creating table: schemata_config\n";
		$this->execute($sql);


$sql = "
CREATE TABLE `column_sequences` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `schema_id` int(11) NOT NULL,
  `sequence_type` enum('shard_column','sequence') NOT NULL DEFAULT 'shard_column',
  `sequence_name` varchar(50) NOT NULL,
  `next_value` bigint(20) DEFAULT NULL,
  `datatype` enum('string','integer') NOT NULL DEFAULT 'integer',
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `sequence_name` (`sequence_name`,`sequence_type`),
  KEY `next_value` (`next_value`),
  KEY `schema_id` (`schema_id`),
  CONSTRAINT `column_sequences_ibfk_1` FOREIGN KEY (`schema_id`) REFERENCES `schemata` (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=5 DEFAULT CHARSET=latin1;
";

		echo "Creating table: column_sequences\n";
		$this->execute($sql);

$sql = "
CREATE TABLE `shard_map` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `column_id` int(11) NOT NULL,
  `key_value` bigint(20) DEFAULT NULL,
  `key_string_value` varchar(255) DEFAULT NULL,
  `shard_id` int(11) NOT NULL,
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `column_id` (`column_id`,`key_value`,`key_string_value`),
  KEY `key_string_value` (`key_string_value`,`column_id`),
  KEY `shard_id` (`shard_id`),
  CONSTRAINT `shard_map_ibfk_1` FOREIGN KEY (`shard_id`) REFERENCES `shards` (`id`),
  CONSTRAINT `shard_map_ibfk_2` FOREIGN KEY (`column_id`) REFERENCES `column_sequences` (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=10 DEFAULT CHARSET=latin1;
";
		echo "Creating table: shard_map\n";
		$this->execute($sql);

$sql = "
CREATE TABLE `gearman_function_names` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `function_name` varchar(50) NOT NULL,
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `function_name` (`function_name`)
) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=latin1;
";
		echo "Creating table: gearman_function_names\n";
		$this->execute($sql);

$sql = "
CREATE TABLE `gearman_functions` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `schema_id` int(11) NOT NULL,
  `worker_count` int(10) unsigned NOT NULL,
  `is_enabled` tinyint(4) NOT NULL DEFAULT '1',
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  `function_name_id` int(11) NOT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `schema_worker_unique` (`schema_id`,`function_name_id`),
  KEY `fk_gearmand_functions_schema_id_idx` (`schema_id`),
  KEY `gearman_functions_ibfk_1` (`function_name_id`),
  CONSTRAINT `gearman_functions_ibfk_1` FOREIGN KEY (`function_name_id`) REFERENCES `gearman_function_names` (`id`),
  CONSTRAINT `fk_gearmand_functions_schema_id` FOREIGN KEY (`schema_id`) REFERENCES `schemata` (`id`) ON DELETE NO ACTION ON UPDATE NO ACTION
) ENGINE=InnoDB AUTO_INCREMENT=12 DEFAULT CHARSET=latin1;
";
		echo "Creating table: gearman_functions\n";
		$this->execute($sql);

$sql = "
CREATE TABLE `gearman_job_servers` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `schema_id` int(11) NOT NULL,
  `hostname` varchar(50) NOT NULL,
  `port` smallint(5) unsigned NOT NULL,
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `schema_id_hostname_port_unique` (`schema_id`,`hostname`,`port`),
  KEY `schema_id` (`schema_id`),
  CONSTRAINT `gearman_job_servers_ibfk_1` FOREIGN KEY (`schema_id`) REFERENCES `schemata` (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=22 DEFAULT CHARSET=latin1;
";
		echo "Creating table: gearman_job_servers\n";
		$this->execute($sql);

#FIXME: structure will change
$sql = "
CREATE TABLE `gearman_workers` (
  `pid` int(11) NOT NULL,
  `function_id` int(11) DEFAULT NULL,
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  PRIMARY KEY (`id`),
  UNIQUE KEY `function_id` (`function_id`,`pid`),
  CONSTRAINT `gearman_workers_ibfk_1` FOREIGN KEY (`function_id`) REFERENCES `gearman_functions` (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;
";

		echo "Creating table: gearman_workers\n";
		$this->execute($sql);

$sql = "
create table jobs(
  id bigint auto_increment primary key,
  shard_id int not null comment 'shard where job result tables are stored',
  resultset_jobs int not null default 0,
  function_jobs int not null default 0,
  tables_used text not null,
  job_status enum('starting', 'running', 'completed', 'error') default 'start' not null,
  job_result_sql text not null,
  foreign key(shard_id) references shards(id),
  unique(query_id)
);
";

		echo "Creating table: job_status\n";
		$this->execute($sql);

$sql = "
create table job_worker_status(
  id bigint auto_increment primary key,
  job_id bigint not null,
  worker_type enum('store_resultset','function'),
  completion_type enum('ok','error')
  foreign key(job_id) references jobs(id)
);
";

		echo "Creating table: job_status\n";
		$this->execute($sql);

	}

    public function register_job($shard_id, $resultset_jobs, $function_jobs, $tables_used, $job_result_sql) {
        $job_result_sql = $this->escape($job_result_sql);
        $tables_used = $this->escape($tables_used);
        $sql = "insert into jobs (id, shard_id, resultset_jobs, function_jobs, tables_used, job_result_sql)";
        $sql .= "VALUES ( NULL, $shard_id, $resultset_jobs, $function_jobs, '$tables_used', '$job_result_sql')";

        $this->execute($sql);
        $id = $this->my_insert_id();

        return $id;
    }

    public function update_job($job_id, $res_count, $func_count, $tables, $sql) {
        $sql= $this->escape($sql);
        $tables= $this->escape($tables);
        if(!is_numeric($job_id)) return false;
        if(!is_numeric($res_count)) return false;
        if(!is_numeric($func_count)) return false;

        $sql = "update jobs set resultset_jobs = $res_count, function_jobs = $func_count, tables_used = '$tables', job_result_sql = '$sql' where id = $job_id;";
        return $this->execute($sql);
    }

	public function get_params($schema_name = null) {

		if(isset($schema_name)) {
			$subquery = "(select id from schemata where schema_name='" . $this->escape($schema_name) . "' limit 1)";
		} else {
			$subquery = "(select id from schemata where is_default_schema = true limit 1)";
		}
	
		$params = array();	
		$sql = "select var_name, var_value from schemata_config where schema_id = $subquery";
		$stmt = $this->execute($sql);
		while($row = $this->conn->my_fetch_assoc($stmt)) {
			$params[$row['var_name']] = $row['var_value'];	
		}
		return $params;

	}
	
	public function get_shards($schema_name = null, $coord_only = false) {
		if(isset($this->shard_cache[isset($schema_name) ? $schema_name : '.'])) return $this->shard_cache[isset($schema_name) ? $schema_name : '.'];
		if(!isset($schema_name)) {
			$sql = "SELECT shards.*,username as user FROM shards join schemata on schema_id = schemata.id where is_default_schema = true";
		} else {
			$sql = "SELECT shards.*,username as user FROM shards join schemata on schema_id = schemata.id where schema_name = '" . $this->escape($schema_name) . "'";

		}

		if($coord_only) $sql .= " AND coord_shard = true";

		$stmt = $this->execute($sql);
	
		$shards = array();
		while($row = $this->conn->my_fetch_assoc($stmt)) {
			$shard = array();
			foreach($row as $key => $val) {
				if(is_numeric($key)) continue;
				$shard[$key] = $val;
			}
				
			$shards[$row['shard_name']] = $shard;
		}
	
		$this->shard_cache[isset($schema_name) ? $schema_name : '.'] = $shards;

		return $shards;
	}
	
	public function add_shard($shard_name, $shard, $coord_shard = true, $accepts_new_rows = true, $extra_info = "NULL", $schema_name = null) {
                if(empty($shard['user'])) $username = "NULL"; else $username="'" . $this->escape($shard['user']) . "'";
                if(empty($shard['db'])) $db = "NULL"; else $db="'" . $this->escape($shard['db']) . "'";
                if(empty($shard['password'])) $password = "NULL"; else $password="'" . $this->escape($shard['password']) . "'";
                if(empty($shard['port'])) $port = "NULL"; else $port="'" . $this>escape($shard['port']) . "'";
                if(empty($shard['host'])) $host = "NULL"; else $host="'" . $this->escape($shard['host']) . "'";
                if(empty($shard['shard_rdbms'])) $shard_rdbms = "'mysql'"; else $shard_rdbms="'" . $shard['shard_rdbms'] . "'";

		if(isset($schema_name)) {
			$subquery = "(select id from schemata where schema_name='" . $this->escape($schema_name) . "' limit 1)";
		} else {
			$subquery = "(select id from schemata where is_default_schema = true limit 1)";
		}

		$shard_name = $this->escape($shard_name);
		if($extra_info != "NULL" )$extra_info = $this->escape($extra_info);
		$coord_shard = $this->escape($coord_shard);
		

                $sql = "INSERT INTO shards (schema_id, shard_name, extra_info, shard_rdbms, coord_shard, accepts_new_rows, username, password, host, port, db) VALUES(";
                $sql .= "$subquery, '$shard_name',$extra_info,$shard_rdbms,'$coord_shard','$accepts_new_rows','$username',$password,$host,$port,$db)";

                $this->execute($sql);

		$this->refresh_shard_cache($schema_name);
	}

	public function refresh_shard_cache($schema_name = null) {
		$this->shard_cache[isset($schema_name) ? $schema_name : '.' ] = null;
		$this->get_shards($schema_name);
		return true;
	}

	public function remove_shard($shard_name, $schema_name = null) {
		if(isset($schema_name)) {
			$schema_name = $this->escape($schema_name);
			$sql = "delete shards.* from shards join schemata on schema_id = schemata.id where schema_name = '$schema_name' and shard_name = '$shard_name'";
		} else {
			$sql = "delete shards.* from shards join schemata on schema_id = schemata.id where is_default_schema = true and shard_name = '$shard_name'";
		}
		$this->execute($sql);
		$this->refresh_shard_cache($schema_name);
		return true;
	}
	
	public function drop_store() {
		$sql = "DROP DATABASE IF EXISTS " . $this->config['db'];
		$this->execute($sql);
		return true;
	}

	public function add_schema($schema_name, $is_default_schema = 1) {

		#this will throw an exception if the unique key is violated	
		$sql = "insert into schemata (schema_name, is_default_schema) values('" . $this->escape($schema_name) . "', $is_default_schema)";
		$this->execute($sql);

		return true;
	
	}

	public function add_column($column, $schema_name = null,$datatype = 'integer',$next_value=1) {

		if(isset($schema_name)) {
			$subquery = "(select id from schemata where schema_name='" . $this->escape($schema_name) . "' limit 1)";
		} else {
			$subquery = "(select id from schemata where is_default_schema = true limit 1)";
		}

		$sql =  "INSERT INTO column_sequences (schema_id, sequence_name, next_value, datatype) VALUES (";
		$sql .= "$subquery,'$column','$next_value','$next_value','$datatype')";

		$this->execute($sql);

		return true;

	}

	public function move_key($column, $key, $tables, $dest_shard, $schema_name = null ) {
		throw new Exception('Not yet implemented.');
	}

	public function lock_key($column, $key, $lock_mode, $schema_name = null) {
		throw new Exception('Not implemented.');
	}

	public function unlock_key($column, $key, $schema_name = null) {
		throw new Exception('Not implemented.');
	}

	public function get_column_info($column_name, $schema_name = null) {

		$column = $this->escape($column);

		if(isset($schema_name)) {
			$subquery = "(select id from schemata where schema_name='" . $this->escape($schema_name) . "' limit 1)";
		} else {
			$subquery = "(select id from schemata where is_default_schema = true limit 1)";
		}

		$sql = "select * from column_sequences where schema_id = $subquery and sequence_name='$column' and sequence_type = 'shard_column' FOR UPDATE";
		$stmt = $this->execute($sql);

		$row = $this->conn->my_fetch_assoc($stmt);
		if(!$row) throw new Exception('Specified column is not a shard key in the given schema');

		return $row;

	}


	public function new_key($column, $key = null, $shard_name = null, $schema_name = null, $col_info = null) {
		if(isset($schema_name)) {
			$subquery = "(select id from schemata where schema_name='" . $this->escape($schema_name) . "' limit 1)";
		} else {
			$subquery = "(select id from schemata where is_default_schema = true limit 1)";
		}

		if(!isset($col_info)) $col_info = $this->get_column_info($column, $schema_name);		

		if(isset($key)) {

			if($col_info['datatype'] != 'string') {
				if($col_info['next_value'] < $key) {
					$sql = "INSERT INTO column_sequences (schema_id, sequence_name,next_value) values (";
					$sql .= "$subquery, '$column', 1) ON DUPLICATE KEY UPDATE next_value = last_insert_id($key)";
				} 
			}

		} else {

			#increment non-string columns.  this query won't do anything on string columns (0 rows returned by select for string columns)
			$sql = "INSERT INTO column_sequences (schema_id, sequence_name,next_value) VALUES(";
			$sql .= "$subquery, '$column', 1) ";
			$sql .= "ON DUPLICATE KEY UPDATE next_value = last_insert_id(next_value+1)";
			$this->execute($sql);
			$key = $this->my_insert_id();

		}

		
		if(!isset($shard_name)) {
			$sql = "select id, shard_name from shards where schema_id = $subquery order by rand() limit 1";
		} else {
			$sql = "select id, shard_name from shards where schema_id = $subquery and shard_name = '$shard_name' limit 1";
		}
		$stmt = $this->execute($sql);
		$row = $this->conn->my_fetch_assoc($stmt);
		if(!$row) throw new Exception('Could not locate the shard for the row');

		$shard_id = $row['id'];
		$shard_name = $row['shard_name'];

		if($row['datatype'] != 'string') {
			$sql = "INSERT INTO shard_map (column_id, shard_id, key_value) VALUES (";
			$sql .= "$shard_id_sq, $subquery, $key);";
		} else {
			$sql = "INSERT INTO shard_map (column_id, shard_id, key_string_value) VALUES (";
			$sql .= "$shard_id_sq, $subquery, $key);";
		}

		$this->execute($sql);
		        
		return array($shard_name => $id);


	}

	#returns an array which maps shard names to id values
	#will create the mapping if it does not exist

	/*
	output:
	$map[] = array('shard1' => 1), array('shard20', 5), ...

	You can map with LIKE, IN, equals(=) or null safe equals(<=>).  
	$key should be an IN list "(1,2,3,4,5)" if the operator is IN

	*/
	public function &map($column, $key, $schema_name=null, $operator = "=", $col_info  = null, $create_on_miss = true) {

		if(!isset($col_info)) $col_info = $this->get_column_info($column, $schema_name);		

		if($col_info['datatype'] != 'string') {
			if($operator == "=" || $operator == "<=>") {
				if(!is_numeric($key)) throw new Exception('Tried to map a string value to an integer mapping');
			} 
			$sql = "select shard_name,key_value as val from shard_map join shards on shard_id = shards.id where column_id = {$col_info['id']} and key_value {$operator} {$key}";
		} else {
			$sql = "select shard_name,key_string_value as val from shard_map join shards on shard_id = shards.id where column_id = {$col_info['id']} and key_string_value {$operator} {$key}";
		}

		$stmt = $this->execute($sql);	

		$map = array();

		while($row = $this->my_fetch_assoc($stmt)) {
			if(empty($map[$row['shard_name']])) $map[$row['shard_name']] = array();
			$map[$row['shard_name']] = $row['val'];
		}
		

		if(empty($map)) {

			if($create_on_miss) {
				#map to a random shard
				$info = $this->new_key($column, $key, null, $schema_name, $col_info);
				return $info;
			} 

			return false;
			
		}

		return($map);
	
	}

	#get all the keys which belong to a particular shard
	public function &get_keys($column, $shard_name, $schema_name = null, $col_info = null) {
		
		if(!isset($col_info)) $col_info = $this->get_column_info($column, $schema_name);

		if($col_info['datatype'] != 'string') {
			$sql = "select key_value val from shard_map join shards on shards_id = shards.id where shard_name = '$shard_name' and column_id = {$col_info['id']}";
		} else {
			$sql = "select key_string_value val from shard_map join shards on shards_id = shards.id where shard_name = '$shard_name' and column_id = {$col_info['id']}";
		}
		$stmt = $this->execute($sql);	
		$keys = array();

		while($row = $this->my_fetch_assoc($stmt)) {
			$keys[] = $row['val'];
		}
	
		return $keys;
	}

	public function &get_key_distribution($column, $schema_name = null) {
		
		if(isset($schema_name)) {	
			$schema_name = $this->escape($schema_name);
			$sql = "select shard_name, count(*) cnt from shard_map join shards on shard_id = shards.id join column_sequences on column_id = column_sequences.id where column_name = '$column' and sequence_type = 'shard_column' and schema_name = '$schema_name' GROUP BY shard_name";
		} else {
			$sql = "select shard_name, count(*) cnt from shard_map join shards on shard_id = shards.id join column_sequences on column_id = column_sequences.id where column_name = '$column' and sequence_type = 'shard_column' and is_default_schema = true GROUP BY shard_name";

		}

		$stmt = $this->execute($sql);
		while($row = $this->my_fetch_assoc($stmt)) {
			$rows[$row['shard_name']] = $row['cnt'];
		}
	
		return $rows;
	}


	public function get_column($schema_name = null) {
		if(!isset($schema_name)) {
			$sql = "select column_name from shard_columns join schemata on schema_id = schemata.id where is_default_schema = true";
		} else {
			$sql = "select column_name from shard_columns join schemata on schema_id = schemata.id where schema_name = '$schema_name'";
		}
		$stmt = $this->execute($sql);	
		while( $row = $this->conn->my_fetch_assoc($stmt) ) {
			$column = $row['column_name'];
		}
		return $column;
	}

	public function get_gearman_servers($schema_name = null) {
		if(isset($schema_name)) {
			$subquery = "(select id from schemata where schema_name='" . $this->escape($schema_name) . "' limit 1)";
		} else {
			$subquery = "(select id from schemata where is_default_schema = true limit 1)";
		}

		$sql = "select concat(hostname, ':', port) as server from gearman_job_servers where schema_id = $subquery and enabled = 1";
		$stmt=$this->execute($sql);
		$servers = array();
		while( $row = $this->conn->my_fetch_assoc($stmt) ) {
			$servers[] = $row['server'];
		}
		return $servers;

	}
	
}

class HashShardKeyMapper extends DirectoryShardKeyMapper{

	public function __construct($config, $no_connect = false ) {
		parent::__construct($config);
	} 

	public function remove_shard($shard_id, $schema = null) {
		throw new Exception('Can not remove shard: Removing a shard would change the hash distribution');
	}
	
	public function move_key($column, $key, $tables, $dest_shard, $schema_name = null ) {
		throw new Exception('Hash partitions are fixed.');
	}

	public function lock_key($column, $key, $lock_mode, $schema = null) {
		throw new Exception('Locks are not supported.');
	}

	public function unlock_key($column, $key, $schema = null) {
		throw new Exception('Locks are not supported.');
	}

	#get a new key value for the given partition
	public function new_key($column, $key = null, $shard_name = null, $schema_name = null, $col_info = null) {
		throw new Exception('Hash partitioning does not support key generation');
	}

	#get all the keys which belong to a particular shard
	public function &get_keys($column, $shard_name, $schema_name = null, $col_info = null) {
		throw new Exception('Hash partitioning does not use a data store and can not enumerate keys');
	}

	public function &get_key_distribution($column, $schema_name = null) {
		throw new Exception('Hash partitioning does not use a data store and can not compute the distribution of keys');
	}

	#returns a shard object
        public function &map($column, $key, $schema_name=null, $operator = "=", $col_info  = null, $create_on_miss = true) {
        static $null = null;
		static $cache = array();
		$shards = $this->get_shards($schema_name);

		if($operator != '=' && strtolower($operator) != 'in') return $null;

		if(strtolower($operator) == "in") {
			$vals = explode(',', trim($key,'()'));
			$prev_item = "";
			$list = array();
			foreach($vals as $item) {
				if($prev_item != "") $item = $prev_item . $item;
				#did explode split a string containing an embedded comma?
				if(!is_numeric($item) && !in_array(substr(trim($item), -1), array("'",'"'))) {
					$prev_item = $item;
				} else {
					$prev_item = "";
				}

				$list[] = $item;
			}

		} else {
			$list = array($key);
		}


		$map = array();
		$shard_names = array_keys($shards);
		foreach($list as $item) {

			if(is_numeric($item)) {
				$map[] = array($shard_names[ $item % count($shard_names) ] => $item);
			} else {
				$map[] = array($shard_names[ (abs(crc32($item)) % count($shard_names)) ] => $item);
			}

		}

		return $map;
	}



}
